package com.example.sendmessage.server;

import com.example.sendmessage.common.Consumer;
import com.example.sendmessage.common.MqException;
import com.example.sendmessage.server.core.*;
import com.example.sendmessage.server.dao.DiskDataCenter;
import com.example.sendmessage.server.dao.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost {
    private String virtualName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    // 添加锁对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    public VirtualHost(String virtualName) {
        this.virtualName = virtualName;
        try {
            diskDataCenter.init();
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MqException | ClassNotFoundException | IOException e) {
            System.out.println("[VirtualHost] 初始化失败！VirtualName = " + virtualName);
            e.printStackTrace();
        }
    }

    /**
     * 虚拟机添加一个交换机
     * @param exchangeName 添加交换机的名字
     * @param type 交换机类型
     * @param durable 是否持久化
     * @param autoDelete 是否自动删除
     * @param arguments 其他参数
     * @return 是否创建成功
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable,
                                boolean autoDelete, Map<String, Object> arguments) {
        // 逻辑隔离，让每个虚拟机内的其他参数隔离
        exchangeName = virtualName + exchangeName;
        try {
            synchronized(exchangeLocker) {
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在，不能重复创建！exchangeName = " + exchangeName);
                    return true;
                }
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(type);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                if (durable) {
                    // 持久化
                    diskDataCenter.insertExchange(exchange);
                    System.out.println("[VirtualHost] 交换机持久化成功！exchangeName = " + exchangeName);
                }
                // 必须写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机添加成功！exchangeName = " + exchangeName);
                return true;
            }
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机添加失败！exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除交换机
     * @param exchangeName 删除交换机的名字
     * @return 删除结果
     */
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange == null) {
                    throw new MqException("[VirtualHost] 交换机不存在无法删除！");
                }
                if (exchange.isDurable()) {
                    // 删除持久化数据
                    diskDataCenter.deleteExchange(exchangeName);
                    System.out.println("[VirtualHost] 交换机持久化数据删除成功！exchangeName = " + exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！exchangeName = " + exchangeName);
            }
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败！exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 创建队列
     * @param queueName 交换机名字
     * @param durable 是否持久化
     * @param exclusive 是否专用
     * @param autoDelete 是否自动删除
     * @param arguments 其他参数
     * @return 创建的结果
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String, Object> arguments) {
        queueName = virtualName + queueName;
        try {
            synchronized(queueLocker) {
                Queue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在！queueName = " + queueName);
                    return true;
                }
                Queue queue = new Queue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！queueName = " + queueName);
                return true;
            }
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除交换机
     * @param queueName 待删除交换机的名字
     * @return 删除结果
     */
    public boolean queueDelete(String queueName) {
        queueName = virtualName + queueName;
        try {
            Queue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 队列不存在，无法删除！queueName = " + queueName);
            }
            if (queue.isDurable()) {
                // 删除持久化数据
                diskDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列持久化数据删除成功！queueName = " + queueName);
            }
            memoryDataCenter.deleteQueue(queueName);
            System.out.println("[VirtualHost] 队列删除成功！queueName = " + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 绑定队列与交换机
     * @param exchangeName 绑定的交换机名
     * @param queueName 绑定的队列名
     * @param bindingKey 绑定关键字
     * @return 是否绑定成功
     */
    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        exchangeName = virtualName + exchangeName;
        queueName = virtualName + queueName;
        try {
            // 加锁
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 查询绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("绑定已经存在，不能重复绑定！exchangeName = " + exchangeName +
                                "，queueName = " + queueName);
                    }
                    // 判断绑定关系是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] 绑定关键字非法！bindingKey = " + bindingKey);
                    }
                    // 创建绑定
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 查询交换机和队列是否存在
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName = " + exchangeName);
                    }
                    Queue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName);
                    }
                    // 判断是否持久化
                    if (exchange.isDurable() && queue.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功！exchangeName = " + exchangeName +
                            "，queueName = " + queueName);
                    return true;
                }
            }
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列绑定失败！exchangeName = " + exchangeName +
                    "，queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 解除交换机与队列的绑定
     * @param exchangeName 解除绑定的交换机名称
     * @param queueName 解除绑定的队列名称
     * @return 是否解绑成功
     */
    public boolean queueUnBind(String exchangeName, String queueName) {
        exchangeName = virtualName + exchangeName;
        queueName = virtualName + queueName;
        try {
            // 进行加锁操作，防止多线程出现异常
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 绑定不存在，无法解绑！exchangeName = " + exchangeName +
                                "，queueName = " + queueName);
                    }
                    // 不用判断交换机和队列是否存在
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                    return true;
                }
            }
        } catch (Exception e) {
            System.out.println("[VirtualHost] 解除绑定失败！exchangeName = " + exchangeName +
                    "，queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 基本推送功能
     * @param exchangeName 推送的交换机名称
     * @param routingKey 推送的路由键
     * @param basicProperties 这里用于设置消息的持久化
     * @return 是否推送成功
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        // 虚拟名称
        exchangeName = virtualName + exchangeName;
        try {
            // 获取交换机
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName = " + exchangeName);
            }
            // 检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey非法！routingKey = " + routingKey);
            }
            // 如果是直连交换机，则创建一个队列
            if (exchange.getType() == ExchangeType.DIRECT) {
                String queueName = virtualName + routingKey;
                Queue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName);
                }
                // 发送消息
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                sendMessage(queue, message);
            } else {
                // Fanout Topic 交换机
                // 查看该队列所有的绑定关系，通过绑定关系进行转发
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    Binding binding = entry.getValue();
                    Queue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 可能是队列已经删除了，不影响
                        System.out.println("[VirtualHost] 队列不存在！queueName = "+ binding.getQueueName());
                        continue;
                    }
                    // 创建消息
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 根据绑定的关键字进行匹配推送
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息推送失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(Queue queue, Message message) {
        try {
            // 判断消息是否需要持久化
            if (message.getDeliverMode() == 2) {
                diskDataCenter.sendMessage(queue, message);
            }
            // 将消息推送到队列
            memoryDataCenter.sendMessage(queue, message);
            // 通知消费者可以消费消息了
            consumerManager.notifyConsume(queue.getName());
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息推送失败！");
            e.printStackTrace();
        }
    }

    /**
     * 基础消费功能，消费者需要实现Consumer接口
     * @param consumerTag 消费者id
     * @param queueName 消费的队列名
     * @param autoAck 是否自动确认应答
     * @param consumer 回调函数
     * @return 是否消费成功
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        queueName = virtualName + queueName;
        try {
            // 给对应的队列添加消费者
            consumerManager.addConsumer(consumerTag, queueName, autoAck,consumer);
            System.out.println("[VirtualHost] 消费者添加成功！queueName = " + queueName + ", consumerTag = " + consumerTag);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息消费失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 基础消息确认功能
     * @param queueName 确认消息队列名
     * @param messageId 确认消息的id
     * @return 是否确认成功
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualName + queueName;
        try {
            Queue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 队列不存在！queueName = "+ queueName);
            }
            // 确认消息是否存在
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 消息不存在！messageId = " + messageId);
            }
            // 删除消息
            if (message.getDeliverMode() == 2) {
                // 删除持久化消息
                diskDataCenter.deleteMessage(queue, message);
            }
            // 删除内存消息
            memoryDataCenter.deleteMessage(messageId);
            // 删除确认消息队列中的消息
            memoryDataCenter.deleteMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] 消息确认成功！queueName = " + queueName + ", messageId = " + messageId);
            return true;
        } catch(Exception e) {
            System.out.println("[VirtualHost] 消息确认失败！queueName = " + queueName + ", messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }


    // 给消费管理提供数据管理支持
    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }
}
